package com.boarsoft.boar.batch.demo.flow.bigfile;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.boarsoft.boar.batch.demo.flow.bigfile.SplitDpfm01.PartFile;
import com.boarsoft.flow.core.bean.ProcessHandler;
import com.boarsoft.hibernate.biz.SimpleBizImpl;

/**
 * @ClassName:  BigFileSplit   
 * @Description:拆分大文件
 * @author: 
 * @date:   2019年2月15日 上午9:50:59   
 * @Copyright: 2019
 */
public class SplitDpfm02 extends SimpleBizImpl implements ProcessHandler {

	private static Logger log = LoggerFactory.getLogger(SplitDpfm02.class);

	private static final long originFileSize = 1024 * 1024 * 200;// 拆分大于originFileSize文件
	private static final int blockFileSize = 1024 * 1024 * 128;// 防止中文乱码，必须取2的N次方，按此长度拆分

	private static final char jxSeparator = '\03';

	public Object process(String entry, Object data, Throwable throwable) throws Throwable {

		long start = System.currentTimeMillis();
//		String fileName = "c:\\jxpoc\\dpfm02copy.dat";
		String fileName = "/home/app/data/dpfm02copy.dat";
		File sourceFile = new File(fileName);
		List<String> parts = new ArrayList<>();
		if (sourceFile.length() >= originFileSize) {
//			String cvsFileName = fileName.replaceAll("\\\\", "/");
			String cvsFileName = fileName;
			SplitDpfm02 fileSplitUtil = new SplitDpfm02();
			System.out.println("开始拆分dpfm02文件");
			parts = fileSplitUtil.splitBySize(cvsFileName, blockFileSize);
			for (String part : parts) {
				System.out.println("拆分后的文件: " + part);
			}
		}
		System.out.println("总文件长度" + sourceFile.length() + ",拆分文件耗时:" + (System.currentTimeMillis() - start) + "ms.");
		
		Map<String, Object> pm = (Map<String, Object>) data;
		final int commitCount = (Integer) pm.get("commitCount");
		if (commitCount > 1000) {
			throw new Exception("commit lines are too many");
		}
		pm.put("list", parts);
		
		return data;
	}

	/**
	 * @Title: splitBySize   
	 * @Description:  按照文件大小拆分大文件
	 * @param fileName 待拆分的完整文件名
	 * @param byteSize 按多少字节大小拆分
	 * @return 拆分后的文件名列表 
	 * @throws
	 */
	public List<String> splitBySize(String fileName, int byteSize) throws IOException, InterruptedException {

		List<String> parts = new ArrayList<String>();
		File file = new File(fileName);
		int count = (int) Math.ceil(file.length() / (double) byteSize); // 拆分的文件个数
		int countLen = (count + "").length();
		RandomAccessFile rFile = new RandomAccessFile(fileName, "r");
		long totalLen = rFile.length();
		CountDownLatch latch = new CountDownLatch(count);
		//单进程执行
		for (int i = 0; i < count; i++) {
			String partFileName = file.getPath() + "." + leftPad((i + 1) + "", countLen, '0') + ".dat";
			int readSize = byteSize;
			long startPos = (long) i * byteSize;
			long nextPos = (long) (i + 1) * byteSize;
			if (nextPos > totalLen) {
				readSize = (int) (totalLen - startPos);
			}
			new SplitRunnable(readSize, startPos, partFileName, file, latch).run();
			parts.add(partFileName);
		}
		//多进程执行
//		for (int i = 0; i < count; i++) {
//			String partFileName = file.getPath() + "." + leftPad((i + 1) + "", countLen, '0') + ".dat";
//			int readSize = byteSize;
//			long startPos = (long) i * byteSize;
//			long nextPos = (long) (i + 1) * byteSize;
//			if (nextPos > totalLen) {
//				readSize = (int) (totalLen - startPos);
//			}
////			System.out.println("即将执行的线程文件名 : " + partFileName);
//			SplitRunnable r = new SplitRunnable(readSize, startPos, partFileName, file, latch);
//	        Thread t = new Thread(r);
//	        t.start();
//	        parts.add(partFileName);
//		}
		latch.await();// 等待所有文件写完
		// 由于切割时可能会导致行被切断，加工所有的的分割文件,合并行
		rFile.close();
		mergeRow(parts);

		// 加工后的文件统一命名
		for (int i=0; i<parts.size(); i++) {
			 String part = parts.get(i) +".bak";
			 parts.set(i, part);
		}
		
		return parts;
	}

	/**
	 * @Title: SplitRunnable   
	 * @Description:  分割文件处理类
	 */
	private class SplitRunnable implements Runnable {
		int byteSize;
		String partFileName;
		File originFile;
		long startPos;
		CountDownLatch latch;

		public SplitRunnable(int byteSize, long startPos, String partFileName, File originFile, CountDownLatch latch) {
			this.startPos = startPos;
			this.byteSize = byteSize;
			this.partFileName = partFileName;
			this.originFile = originFile;
			this.latch = latch;
		}

		public void run() {
			RandomAccessFile rFile;
			OutputStream os;
			try {
//				System.out.println("本线程处理的文件名：" + partFileName);
//				System.out.println("本线程ID：" + Thread.currentThread().getId());
				rFile = new RandomAccessFile(originFile, "r");
				byte[] b = new byte[byteSize];
				rFile.seek(startPos);// 移动指针到每“段”开头
				int s = rFile.read(b);
				os = new FileOutputStream(partFileName);
				os.write(b, 0, s);
				os.flush();
				os.close();
				latch.countDown();
			} catch (IOException e) {
				log.error(e.getMessage());
				System.out.println("run里面的错误");
				latch.countDown();
			}
		}
	}

	/**
	 * @Title: mergeRow   
	 * @Description:  合并被切断的行
	 * @param parts 分割后的文件列表
	 * @return  
	 * @throws
	 */
	private void mergeRow(List<String> parts) {
		List<PartFile> partFiles = new ArrayList<PartFile>();
		try {
			// 组装被切分表对象
			for (int i = 0; i < parts.size(); i++) {
				String partFileName = parts.get(i);
				File splitFileTemp = new File(partFileName);
				if (splitFileTemp.exists()) {
					PartFile partFile = new PartFile();
					BufferedReader reader = new BufferedReader(
							new InputStreamReader(new FileInputStream(splitFileTemp), "gbk")); // gbk是中文
					String firstRow = reader.readLine();
					String secondRow = reader.readLine();
					String endRow = readLastLine(partFileName);
					partFile.setPartFileName(partFileName);
					partFile.setFirstRow(firstRow);
					partFile.setEndRow(endRow);
					if (i >= 1) {
						String prePartFile = parts.get(i - 1);
						String preEndRow = readLastLine(prePartFile);
//						partFile.setFirstIsFull(getCharCount(firstRow + preEndRow) > getCharCount(secondRow));
						partFile.setFirstIsFull(false);
						partFile.setPart(getCharCount(preEndRow) < getCharCount(secondRow));
					}
					partFiles.add(partFile);
					reader.close();
				}
			}
			// 进行需要合并的行的写入
			System.out.println("拆分文件的个数: " + partFiles.size());
			for (int i = 0; i < partFiles.size() - 1; i++) {
				PartFile partFile = partFiles.get(i);
				PartFile partFileNext = partFiles.get(i + 1);
				StringBuilder sb = new StringBuilder();
//				if (partFileNext.getFirstIsFull()) {
//					sb.append("\r\n");
//					sb.append(partFileNext.getFirstRow());
//				} else {
//					sb.append(partFileNext.getFirstRow());
//				}
				if (partFileNext.isPart()) {
					sb.append(partFileNext.getFirstRow());
					writeLastLine(partFile.getPartFileName(), sb.toString());
				} 
			}
			//第二个文件的第一行合并到第一个文件后，删除第一行
			for (int i = 0; i < partFiles.size(); i++) {
				PartFile partFile = partFiles.get(i);
				DeleteFirstLine(partFile.getPartFileName(), partFile.getFirstRow(), partFile.isPart());
			}

		} catch (Exception e) {
			log.error(e.getMessage());
			System.out.println("merge里面的错误");
		}
	}

	/**
	 * @Title: getCharCount   
	 * @Description:  得到某个字符出现的次数
	 * @param s 字符串
	 * @return  出现次数
	 * @throws
	 */
	private int getCharCount(String s) {
		int count = 0;
		for (int i = 0; i < s.length(); i++) {
			if (s.charAt(i) == jxSeparator) {
				count++;
			}
		}
		return count;
	}

	/**
	 * @Title: getFileRow   
	 * @Description:  采用BufferedInputStream方式读取文件行数
	 * @param filename
	 * @return 行数
	 * @throws
	 */
	public int getFileRow(String filename) throws IOException {
		InputStream is = new BufferedInputStream(new FileInputStream(filename));
		byte[] c = new byte[1024];
		int count = 0;
		int readChars = 0;
		while ((readChars = is.read(c)) != -1) {
			for (int i = 0; i < readChars; ++i) {
				if (c[i] == '\n')
					++count;
			}
		}
		is.close();
		return count;
	}

	/**
	 * @Title: readLastLine   
	 * @Description:  读取最后一行数据
	 * @param filename
	 * @return String 最后一行数据
	 * @throws IOException
	 */
	private String readLastLine(String filename) throws IOException {
		// 使用RandomAccessFile , 从后找最后一行数据
		RandomAccessFile raf = new RandomAccessFile(filename, "r");
		long len = raf.length();
		String lastLine = "";
		if (len != 0L) {
			long pos = len - 1;
			while (pos > 0) {
				pos--;
				raf.seek(pos);
				if (raf.readByte() == '\n') {
					lastLine = raf.readLine();
					lastLine = new String(lastLine.getBytes("8859_1"), "gbk");
					break;
				}
			}
		}
		raf.close();
		return lastLine;
	}

	/**
	 * @Title: writeLastLine   
	 * @Description:  修改最后一行数据
	 * @param filename
	 * @param lastString 需要写入的数据
	 * @return 
	 * @throws IOException
	 */
	private void writeLastLine(String fileName, String lastString) {
		try {
			// 打开一个随机访问文件流，按读写方式
			RandomAccessFile randomFile = new RandomAccessFile(fileName, "rw");
			// 文件长度，字节数
			long fileLength = randomFile.length();
			// 将写文件指针移到文件尾。
			randomFile.seek(fileLength);
			// 此处必须加gbk，否则会出现写入乱码
			randomFile.write(lastString.getBytes("gbk"));
			randomFile.close();
//			new File(fileName).renameTo(new File(fileName + ".bak"));  
		} catch (IOException e) {
			log.error(e.getMessage());
			System.out.println("write里面的错误");
		}
	}
	
	/**
	 * @Title: DeleteFirstLine   
	 * @Description:  删除第一行数据
	 * @param filename
	 * @param firstLine 需要删除的数据
	 * @return 
	 * @throws IOException
	 */
	private void DeleteFirstLine(String fileName, String firstLine, boolean isPart) {
		try {
			// 打开一个随机访问文件流，按读方式
			RandomAccessFile randomFile = new RandomAccessFile(fileName, "r");
			long lenFile = randomFile.length();
			long lenLine = firstLine.length();
			byte[] b = new byte[(int)lenFile];
			//删除第一行后，新产生一个文件
			String partFileName = fileName + ".bak";
			if (!isPart) {
				randomFile.seek(0);
			} else {
				// 将读文件指针移到第一行尾。
				randomFile.seek(lenLine + 1);
			}
			int s = randomFile.read(b);
			OutputStream os = new FileOutputStream(partFileName);
			os.write(b, 0, s);
			os.flush();
			os.close();
			randomFile.close();
			new File(fileName).delete();
		} catch (IOException e) {
			log.error(e.getMessage());
			System.out.println("delete里面的错误");
		}
	}

	/**
	 * @Title: leftPad   
	 * @Description:  左填充
	 * @param str
	 * @param length
	 * @param ch
	 * @return 填充后的字符串
	 * @throws 
	 */
	public static String leftPad(String str, int length, char ch) {
		if (str.length() >= length) {
			return str;
		}
		char[] chs = new char[length];
		Arrays.fill(chs, ch);
		char[] src = str.toCharArray();
		System.arraycopy(src, 0, chs, length - src.length, src.length);
		return new String(chs);
	}

	/**
	 * @Title: PartFile   
	 * @Description:  合并行的内部类，在合并行时使用
	 */
	class PartFile {
		private String partFileName;
		private String firstRow;
		private String endRow;
		private boolean firstIsFull;
		private boolean isPart;

		public String getPartFileName() {
			return partFileName;
		}

		public void setPartFileName(String partFileName) {
			this.partFileName = partFileName;
		}

		public String getFirstRow() {
			return firstRow;
		}

		public void setFirstRow(String firstRow) {
			this.firstRow = firstRow;
		}

		public String getEndRow() {
			return endRow;
		}

		public void setEndRow(String endRow) {
			this.endRow = endRow;
		}

		public boolean getFirstIsFull() {
			return firstIsFull;
		}

		public void setFirstIsFull(boolean firstIsFull) {
			this.firstIsFull = firstIsFull;
		}

		public boolean isPart() {
			return isPart;
		}

		public void setPart(boolean isPart) {
			this.isPart = isPart;
		}
	}

}